// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/base/symbology/product_holder_manager.h"

#include <algorithm>
#include <experimental/filesystem>
#include <string>
#include <tuple>
#include <utility>

#include <sys/syscall.h>
#include <sys/types.h>
#include <unistd.h>

#include "coin2/base/date_util.h"
#include "coin2/base/hostname_util.h"
#include "coin2/exchange/factory/symbology/product.h"

#define CHECK_INTERVAL 10

namespace coin2::exchange::base::symbology {

using coin2::exchange::base::market::ExchangeTypeToString;
using coin2::exchange::base::market::MarketExchangeApi;
using coin2::exchange::base::market::MarketTypeToString;

static constexpr int64_t day = 3600 * 24 * 1'000'000'000LL;
static constexpr int64_t kOldestDate = 1637884800'000'000'000LL;  // 2021-11-26
static constexpr char kFallbackPath[] = "data/coin2/product_info";
static constexpr char kDefaultSearchPath[] =
    "/remote/iosg/home-2/public/bot-coin/coin_product_info/data:"
    "../coin_product_info/data:"
    "../../coin_product_info/data:"
    "~/workspace/coin_product_info/data:"
    "data/coin2/product_info";

namespace {

inline int64_t GetThreadId() { return syscall(__NR_gettid); }

inline int64_t TimestampToUtc0(int64_t ts) { return (ts / day) * day; }

std::vector<std::string> GetSearchPathList(const std::string& paths) {
  auto splitted = Split(paths, ":");
  std::vector<std::string> result;
  for (const auto& path : splitted) {
    if (path[0] == '~') {
      result.push_back(ExpandUser(path));
    } else {
      result.push_back(path);
    }
  }
  return result;
}

}  // namespace

ProductHolderManager::ProductHolderManager(const coin::proto::ProductHolderManagerConfig& config)
    : config_(config) {
  std::vector<std::string> search_path_list;
  if (config_.has_root_dir()) {
    search_path_list = GetSearchPathList(config_.root_dir());
  } else {
    search_path_list = GetSearchPathList(kDefaultSearchPath);
  }
  product_info_constants_.reset(new ProductInfoConstantsManager(search_path_list));
  LOG(INFO) << "product holder manager created! " << this;
}

void ProductHolderManager::SetHotUpdateStatus(bool enable) {
  config_.set_hot_update_enabled(enable);
  product_info_constants_->SetHotUpdateStatus(enable);
}

bool ProductHolderManager::JsonNeedLoad(
    coin::proto::MarketType market,
    coin::proto::ExchangeType exchange,
    int64_t midnight_ts) {
  auto key = std::make_tuple(market, exchange, midnight_ts);
  auto need_load = json_loaded_.count(key) > 0 ? false : true;
  if (!need_load && config_.hot_update_enabled()) {
    if (product_info_constants_->IsPiConstantUpdated(market, exchange, midnight_ts)) {
      need_load = true;
    }
  }

  return need_load;
}

const IProductHolder* ProductHolderManager::GetProductHolder(
    const IProduct& product,
    int64_t timestamp) {
  // Always use latest product holder.
  if (config_.search_policy() == coin::proto::ProductHolderManagerConfig::Latest) {
    timestamp = GetCurrentTimestamp();
  }
  int64_t midnight_ts = TimestampToUtc0(timestamp);
  if (midnight_ts < kOldestDate) {
    LOG(INFO) << "Use 20211126 PI instead!";
    midnight_ts = kOldestDate;
  }
  return GetProductHolderAux(product, timestamp, midnight_ts, config_.fallback_day());
}

const IProductHolder* ProductHolderManager::GetProductHolderAux(
    const IProduct& product,
    int64_t timestamp,
    int64_t midnight_ts,
    int depth) {
  if (midnight_ts < 0 || depth < 0) {
    // char buf[1024];
    //  LOG(ERROR) << product.key_string() << ": product info not found! "
    //             << "phm object: " << this << ", "
    //             << "thread id: " << GetThreadId() << ", "
    //             << "pid: " << getpid() << ", "
    //             << "hostname: " << GetLocalHostname() << ", "
    //             << "cwd: " << getcwd(buf, sizeof(buf)) << ", " << product.market_name() << ":"
    //             << product.exchange_name() << ":" << product.absolute_norm();
    throw std::runtime_error(
        std::string("Cannot found product holder for " + product.absolute_norm()));
  }
  if (midnight_ts < kOldestDate) {
    midnight_ts = 0L;
  }

  if (JsonNeedLoad(product.market(), product.exchange(), midnight_ts)) {
    bool res = LoadProductHolder(product.market(), product.exchange(), timestamp, midnight_ts);
    if (!res) {
      if (depth == 0) {
        throw std::runtime_error(fmt::format(
            "Cannot load product holder for {}-{}-{}",
            MarketTypeToString(product.market()),
            ExchangeTypeToString(product.exchange()),
            product.absolute_norm()));
      } else {
        return GetProductHolderAux(product, timestamp, midnight_ts - day, depth - 1);
      }
    }
  }

  auto key1 =
      std::make_tuple(product.market(), product.exchange(), midnight_ts, product.absolute_norm());
  if (product_holders_cache_.count(key1) > 0) {
    return product_holders_cache_.at(key1).get();
  }

  auto key2 = std::make_tuple(
      product.market(),
      product.exchange(),
      TimestampToUtc0(timestamp),
      product.absolute_norm());
  if (product_holders_cache_.count(key2) > 0) {
    return product_holders_cache_.at(key2).get();
  }

  if (depth == 0) {
    throw std::runtime_error(
        std::string("Cannot found product holder for " + product.absolute_norm()));
  } else {
    //     LOG(INFO) << "Load product info for " << product.market_name() << ":" <<
    //     product.exchange_name()
    //               << ":" << product.absolute_norm() << ", "
    //               << "timestamp: " << timestamp << ", "
    //               << "phm object: " << this << ", "
    //               << "thread id: " << GetThreadId() << ", "
    //               << "pid: " << getpid();
    return GetProductHolderAux(product, timestamp, midnight_ts - day, depth - 1);
  }
}

bool ProductHolderManager::LoadProductHolder(
    coin::proto::MarketType market,
    coin::proto::ExchangeType exchange,
    int64_t timestamp,
    int64_t midnight_ts) {
  // Json file correponds to the key has been loaded.
  auto pi_constants =
      product_info_constants_->GetProductInfoConstant(market, exchange, midnight_ts);
  if (!pi_constants) {
    return false;
  }

  auto key = std::make_tuple(market, exchange, midnight_ts);
  json_loaded_.insert(key);
  for (const auto& pi : pi_constants->GetProductInfoBundle().product_infos()) {
    auto product = CreateProductFromNormString(market, exchange, pi.symbol(), timestamp, false);
    if (!product) continue;
    // int64_t m_ts = TimestampToUtc0(timestamp);
    auto key = std::make_tuple(
        product->market(),
        product->exchange(),
        midnight_ts,
        product->absolute_norm());
    // if midnight_ts == 0, first pi is chosen.
    if (product_holders_cache_.count(key) == 0) {
      product_holders_cache_[key] = std::make_unique<ProductHolder>(*product, pi);
    }
  }
  return true;
}

ProductInfoConstantsManager::ProductInfoConstantsManager(const std::vector<std::string>& paths)
    : files_monitor_(new FileMonitor(CHECK_INTERVAL)), search_path_list_(paths) {
  for (const auto& p : search_path_list_) {
    LOG(INFO) << "search path: " << p;
  }
}

bool ProductInfoConstantsManager::LoadProductInfo(
    const coin::proto::MarketType market,
    const coin::proto::ExchangeType exchange,
    int64_t midnight_ts) {
  using namespace std::experimental::filesystem;
  for (const auto& path : search_path_list_) {
    bool result = LoadProductInfoFromPath(market, exchange, path, midnight_ts);
    if (result) {
      std::string me =
          coin::proto::MarketType_Name(market) + "." + coin::proto::ExchangeType_Name(exchange);
      std::string trading_date = ToDateStr(midnight_ts);
      LOG(INFO) << "Success load product info " << me << ", " << trading_date;
      return true;
    }
  }
  return false;
}

// AddtionalProductInfo:
//   1. list_time
//   2. capture_time
// Other fields values are all ignored.
std::unique_ptr<ProductInfoConstant> ProductInfoConstantsManager::GetAdditionalProductInfo(
    const std::string me) {
  using namespace std::experimental::filesystem;

  std::unique_ptr<ProductInfoConstant> pi;
  for (const auto& search_path : search_path_list_) {
    auto item = search_path;
    item = item.substr(0, item.rfind('/'));
    std::string pattern = (path(item) / path("ico") / path(me)).string() + ".*.json";
    auto files = Glob(pattern);
    if (files.empty()) {
      continue;
    }

    for (const auto& file : files) {
      LOG(INFO) << "Load ico product info file: " << file;
      if (pi) {
        pi->MergeFrom(ProductInfoConstant(file));
      } else {
        pi = std::make_unique<ProductInfoConstant>(file);
      }
    }

    if (pi && pi->GetProductInfoBundle().product_infos_size() > 0) {
      break;
    }
  }

  return pi;
}

bool ProductInfoConstantsManager::LoadProductInfoFromPath(
    const coin::proto::MarketType market,
    const coin::proto::ExchangeType exchange,
    const std::string& search_path,
    int64_t midnight_ts) {
  using namespace std::experimental::filesystem;

  path root_dir(search_path);
  std::string me =
      coin::proto::MarketType_Name(market) + "." + coin::proto::ExchangeType_Name(exchange);
  std::string trading_date = ToDateStr(midnight_ts);

  if (root_dir.string() == std::string(kFallbackPath)) {
    root_dir /= path(me);
  } else {
    root_dir /= path(trading_date);
    if (!exists(root_dir)) {
      return false;
    }
    root_dir /= path(me);
  }

  auto files = Glob(root_dir.string() + ".*.json");
  if (files.empty()) {
    return false;
  }

  auto key = std::make_tuple(market, exchange, midnight_ts);
  auto pi = std::make_unique<ProductInfoConstant>();
  for (const auto& file : files) {
    LOG(INFO) << "Load product info file: " << file;
    pi->MergeFrom(ProductInfoConstant(file));

    if (files_monitor_->HasFile(file)) {
      if (files_monitor_->IsFileUpdated(file)) {
        LOG(INFO) << "Refresh monitor file " << file;
        files_monitor_->UpdateLastChangeTime(file);
      }
    } else if (files_monitor_->AddFile(file)) {
      LOG(INFO) << "Monitor file " << file;
      pi_files_cache_[key].push_back(file);
    } else {
      LOG(ERROR) << "Monitor file err " << file;
    }
  }

  auto ico_pi = GetAdditionalProductInfo(me);
  if (ico_pi) {
    pi->PartialMergeFrom(*ico_pi);
  }

  if (pi_constant_cache_.count(key) > 0) {
    pi_constant_cache_[key]->MergeExtraFrom(*pi);
  } else {
    pi_constant_cache_[key] = std::move(pi);
  }
  return true;
}

const ProductInfoConstant* ProductInfoConstantsManager::GetProductInfoConstant(
    const coin::proto::MarketType market,
    const coin::proto::ExchangeType exchange,
    int64_t timestamp) {
  timestamp = TimestampToUtc0(timestamp);
  if (timestamp < kOldestDate) {
    timestamp = kOldestDate;
  }

  auto key = std::make_tuple(market, exchange, timestamp);
  bool loaded = pi_constant_cache_.count(key) > 0;

  if (loaded) {
    if (hot_update_enabled && IsPiConstantUpdated(market, exchange, timestamp)) {
      LoadProductInfo(market, exchange, timestamp);
    }
  } else {
    bool res = LoadProductInfo(market, exchange, timestamp);
    if (res == false) {
      return nullptr;
    }
  }

  return pi_constant_cache_.at(key).get();
}

bool ProductInfoConstantsManager::IsPiConstantUpdated(
    const coin::proto::MarketType market,
    const coin::proto::ExchangeType exchange,
    int64_t midnight_ts) const {
  auto key = std::make_tuple(market, exchange, midnight_ts);
  auto cur = pi_files_cache_.find(key);
  if (cur == pi_files_cache_.end()) {
    return false;
  }

  const std::list<std::string>& files = cur->second;
  for (const auto& file : files) {
    bool is_updated = files_monitor_->IsFileUpdated(file);
    if (is_updated) {
      return true;
    }
  }

  return false;
}

}  // namespace coin2::exchange::base::symbology
